package com.antu.filegatewaytask.task;


import com.antu.filegatewaytask.service.DemoService;
import com.google.common.collect.Lists;
import org.springframework.stereotype.Component;
import tech.powerjob.common.exception.PowerJobCheckedException;
import tech.powerjob.worker.core.processor.ProcessResult;
import tech.powerjob.worker.core.processor.TaskContext;
import tech.powerjob.worker.core.processor.TaskResult;
import tech.powerjob.worker.core.processor.sdk.MapReduceProcessor;
import tech.powerjob.worker.log.OmsLogger;


import java.util.List;

/**
 * 并行处理器demo
 *
 * @ClassName MapReduceProcessorDemo
 * @Description TODO
 * @Author CYH
 * @Date 2023/12/12 14:57
 */
@Component
public class MapReduceProcessorDemo implements MapReduceProcessor {


    @Override
    public ProcessResult process(TaskContext context) {
        OmsLogger omsLogger = context.getOmsLogger();
        String jobParams = context.getJobParams();
        // root task 负责分发任务
        if (isRootTask()) {
            omsLogger.info("并行处理开始参数为:{}", jobParams);
            omsLogger.info("rootTask开始");
            // 从控制台传递分片参数，假设格式为KV：1,2,3,4
            List<DemoService> subTasks = Lists.newLinkedList();
            for (String num : jobParams.split(",")) {
                subTasks.add(new DemoService(Integer.valueOf(num)));
            }
            try {
                omsLogger.info("开始分发任务");
                map(subTasks, "COMPUTE");
                omsLogger.info("任务分发成功");
            } catch (PowerJobCheckedException e) {
                omsLogger.info("任务分发失败:{}", e.getMessage());
                throw new RuntimeException(e);
            }
            omsLogger.info("root task 执行完毕");
            return new ProcessResult(true, "0");
        }
        //非root task 判断子任务类型
        Object subTask = context.getSubTask();
        if (subTask instanceof DemoService) {
            DemoService service = ((DemoService) subTask);
            // 执行子任务
            omsLogger.info("开始子任务执行参数为:{}", service.getOpNum());
            int res = service.compute();
            omsLogger.info("子任务执行完毕结果为" + res);
            return new ProcessResult(true, String.valueOf(res));
        }
        omsLogger.warn("无法识别的子任务");
        return new ProcessResult(false, "无法识别的子任务" + subTask.getClass().getName());
    }

    @Override
    public ProcessResult reduce(TaskContext context, List<TaskResult> taskResults) {
        OmsLogger omsLogger = context.getOmsLogger();
        omsLogger.info("开始汇总执行结果");
        int sum = 0;
        for (TaskResult tr : taskResults) {
            if (tr.isSuccess()) {
                sum += Integer.parseInt(tr.getResult());
            }
        }
        omsLogger.info("并行处理结束执行结果为{}", sum);
        // 该结果将作为任务最终的执行结果
        return new ProcessResult(true, "并行处理结束结果为:" + sum);

    }


}
